package xin.marcher.wind.migrate.migrate;

import cn.hutool.core.date.DatePattern;
import cn.hutool.core.date.DateTime;
import cn.hutool.core.date.LocalDateTimeUtil;
import com.baomidou.mybatisplus.core.toolkit.CollectionUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.ObjectUtils;
import xin.marcher.framework.common.util.spring.SpringContextUtil;
import xin.marcher.wind.migrate.domain.BinLog;
import xin.marcher.wind.migrate.domain.RangeScroll;
import xin.marcher.wind.migrate.domain.entity.EtlDirtyRecordDO;
import xin.marcher.wind.migrate.domain.entity.EtlProgressDO;
import xin.marcher.wind.migrate.domain.entity.EtlStatisticalDO;
import xin.marcher.wind.migrate.enums.BinlogType;
import xin.marcher.wind.migrate.enums.DBChannel;
import xin.marcher.wind.migrate.enums.EtlProgressStatus;
import xin.marcher.wind.migrate.mapper.MigrateScrollMapper;
import xin.marcher.wind.migrate.service.MigrateService;
import xin.marcher.wind.migrate.util.MigrateCheckUtil;

import java.math.BigDecimal;
import java.util.*;
import java.util.stream.Collectors;

/**
 * 数据核对处理
 */
@Slf4j
public class CheckDataProcessor {

    private static volatile CheckDataProcessor checkDataProcessor;

    private CheckDataProcessor() {

    }

    /**
     * 核验数据
     *
     * @param rangeScroll 要检查的数据抽取模型
     */
    public void checkData(RangeScroll rangeScroll) {
        EtlProgressDO etlProgressDO = addEtlProgress(rangeScroll);
        try {
            //1. 先获取老库的一批数据
            List<Map<String, Object>> sourceList = querySourceList(rangeScroll);
            //2. 再获取新库的一批数据
            List<Map<String, Object>> targetList = queryTargetList(sourceList, rangeScroll);
            //3. 对数据进行核对校验
            Map<BinlogType, List<Map<String, Object>>> comparisonMap = comparison(sourceList, targetList, rangeScroll);
            //4. 对数据进行归正处理
            updateComparisonData(comparisonMap, rangeScroll);
            //5. 完成数据核对校验，更改状态
            updateEtlDirtyRecord(etlProgressDO, EtlProgressStatus.CHECK_SUCCESS.getValue(), rangeScroll, null);
        } catch (Exception e) {
            //数据核对过程失败，只记录数据核对错误信息
            updateEtlDirtyRecord(etlProgressDO, EtlProgressStatus.SUCCESS.getValue(), rangeScroll, e.getMessage());
            log.error("数据核对过程中发生异常 {" + e.getMessage() + "}", etlProgressDO);
        }
    }

    /**
     * 获取到可以核对校验的数据批次集合
     *
     * @return 数据批次集合
     */
    public List<RangeScroll> queryCheckDataList() {
        MigrateScrollMapper migrateScrollMapper = SpringContextUtil.getBean(MigrateScrollMapper.class);
        // 查询已同步的数据
        EtlDirtyRecordDO etlDirtyRecordDO = new EtlDirtyRecordDO();
        etlDirtyRecordDO.setStatus(EtlProgressStatus.SUCCESS.getValue());
        // 默认一次最多查询100条数据
        List<EtlDirtyRecordDO> progressList = migrateScrollMapper.queryDirtyRecordList(etlDirtyRecordDO);
        List<RangeScroll> rangeScrollList = new ArrayList<>();
        if (CollectionUtils.isNotEmpty(progressList)) {
            for (EtlDirtyRecordDO dirtyRecord : progressList) {
                RangeScroll rangeScroll = new RangeScroll();
                rangeScroll.setDomain(dirtyRecord.getDomain());
                rangeScroll.setDomainId(dirtyRecord.getDomainId());
                rangeScroll.setTargetTableName(dirtyRecord.getTargetTableName());
                rangeScroll.setId(dirtyRecord.getId());
                rangeScroll.setTicket(dirtyRecord.getTicket());
                rangeScroll.setCurTicketStage(dirtyRecord.getCurTicketStage());
                rangeScroll.setTableName(dirtyRecord.getLogicModel());
                rangeScroll.setStartScrollId(dirtyRecord.getRecordValue());
                rangeScroll.setPageSize(dirtyRecord.getSyncSize());
                rangeScroll.setEndTime(dirtyRecord.getScrollEndTime());
                rangeScroll.setStartTime(dirtyRecord.getScrollTime());
                rangeScrollList.add(rangeScroll);
            }
        }
        return rangeScrollList;
    }

    /**
     * 构建一个单例模式对象
     *
     * @return CheckDataProcessor实例
     */
    public static CheckDataProcessor getInstance() {
        if (null == checkDataProcessor) {
            synchronized (CheckDataProcessor.class) {
                if (null == checkDataProcessor) {
                    checkDataProcessor = new CheckDataProcessor();
                }
            }
        }
        return checkDataProcessor;
    }

    /**
     * 核对数据不一致的进行修复
     *
     * @param comparisonMap 要比对的数据
     * @param rangeScroll   数据抽取模型
     */
    private void updateComparisonData(Map<BinlogType, List<Map<String, Object>>> comparisonMap, RangeScroll rangeScroll) {
        MigrateService migrateService = SpringContextUtil.getBean(MigrateService.class);

        List<BinLog> updatedList = new ArrayList<>();
        for (Map.Entry<BinlogType, List<Map<String, Object>>> comparisonEntry : comparisonMap.entrySet()) {
            //分别处理需要更新和新增的数据
            for (Map<String, Object> updatedMap : comparisonEntry.getValue()) {
                BinLog binLog = new BinLog();
                binLog.setOperateType(comparisonEntry.getKey().getValue());
                binLog.setTableName(rangeScroll.getTableName());
                binLog.setDataMap(updatedMap);
                updatedList.add(binLog);
            }
        }
        migrateService.migrateBat(rangeScroll, updatedList);
    }

    /**
     * 对数据进行核对比较，并返回核对不一致的数据
     *
     * @param sourceList 源List
     * @param targetList 被比较的List
     * @return 异常数据Map
     */
    private Map<BinlogType, List<Map<String, Object>>> comparison(List<Map<String, Object>> sourceList, List<Map<String, Object>> targetList, RangeScroll rangeScroll) {
        String singleKey = MergeConfig.getSingleKey(rangeScroll.getTableName());
        // 核验数据，主要对比的老库的数据是否和新库不一致，以老库为准,先转换新库的数据对象为MAP，key以唯一标识字段的值为准
        Map<String, Map<String, Object>> targetMap = targetList.stream().collect(Collectors.toMap(
                s -> s.get(singleKey) + "", s -> s
        ));
        Map<BinlogType, List<Map<String, Object>>> errorMap = new HashMap<BinlogType, List<Map<String, Object>>>();
        // 遍历源数据集合信息，验证数据是否匹配
        for (Map<String, Object> sourceMap : sourceList) {
            boolean isComparison = false;
            Object scrollKeyObj = sourceMap.get(singleKey);
            if (scrollKeyObj instanceof String) {
                String scrollKey = (String) scrollKeyObj;
                // 匹配对应的数据
                Map<String, Object> scrollMap = targetMap.get(scrollKey);
                if (null != scrollMap) {
                    isComparison = MigrateCheckUtil.comparison(sourceMap, scrollMap, rangeScroll.getTableName());
                    // 数据未匹配，入异常记录
                    if (isComparison) {
                        //添加需要更新的数据
                        if (errorMap.get(BinlogType.UPDATE) != null) {
                            errorMap.get(BinlogType.UPDATE).add(sourceMap);
                        } else {
                            List<Map<String, Object>> updatedList = new ArrayList<Map<String, Object>>();
                            updatedList.add(sourceMap);
                            errorMap.put(BinlogType.UPDATE, updatedList);
                        }
                    }
                } else {
                    //添加需要新增的数据
                    if (errorMap.get(BinlogType.INSERT) != null) {
                        errorMap.get(BinlogType.INSERT).add(sourceMap);
                    } else {
                        List<Map<String, Object>> insertedList = new ArrayList<Map<String, Object>>();
                        insertedList.add(sourceMap);
                        errorMap.put(BinlogType.INSERT, insertedList);
                    }
                }
            }
        }
        return errorMap;
    }

    /**
     * 根据源数据查询的信息，返回新库的数据信息
     *
     * @param sourceList 新库的数据信息
     */
    private List<Map<String, Object>> queryTargetList(List<Map<String, Object>> sourceList, RangeScroll rangeScroll) {
        MigrateService migrateService = SpringContextUtil.getBean(MigrateService.class);
        List<String> queryList = new ArrayList<>();
        String singleKey = MergeConfig.getSingleKey(rangeScroll.getTableName());
        // 按指定的字段名称，获取批量的具体值作为查询条件
        for (Map<String, Object> map : sourceList) {
            queryList.add(map.get(singleKey) + "");
        }
        // 按组装的集合，查询新库数据对象返回
        return migrateService.findByIdentifiers(rangeScroll, queryList, DBChannel.CHANNEL_2.getValue());
    }

    /**
     * 从源数据，拉取全量同步的某一个批次数据
     *
     * @param rangeScroll 数据抽取模型
     * @return 指定批次数据
     */
    private List<Map<String, Object>> querySourceList(RangeScroll rangeScroll) {
        MigrateService migrateService = SpringContextUtil.getBean(MigrateService.class);
        // 拉取某一个批次的源数据信息
        return migrateService.queryInfoList(rangeScroll);
    }

    /**
     * 初始化记录本次迁移数据
     *
     * @param rangeScroll 数据抽取模型
     * @return 记录本次迁移数据
     */
    private EtlProgressDO addEtlProgress(RangeScroll rangeScroll) {
        MigrateScrollMapper migrateScrollMapper = SpringContextUtil.getBean(MigrateScrollMapper.class);
        // 查询是否已经有核对的数据了
        EtlProgressDO etlProgressDO = migrateScrollMapper.queryEtlProgressCheck(rangeScroll);
        if (null != etlProgressDO) {
            return etlProgressDO;
        }
        etlProgressDO = new EtlProgressDO();
        etlProgressDO.setDomain(rangeScroll.getDomain());
        etlProgressDO.setDomainId(rangeScroll.getDomainId());
        etlProgressDO.setLogicModel(rangeScroll.getTableName());
        etlProgressDO.setTicket(rangeScroll.getTicket());
        etlProgressDO.setCurTicketStage(rangeScroll.getCurTicketStage());
        etlProgressDO.setRetryTimes(0);
        etlProgressDO.setFinishRecord(0);
        etlProgressDO.setProgressType(1);
        etlProgressDO.setStatus(EtlProgressStatus.INIT.getValue());
        //初始化查询的字段值，不存在数据库字段默认设置为1
        if (!ObjectUtils.isEmpty(rangeScroll.getStartScrollId())) {
            etlProgressDO.setScrollId(rangeScroll.getStartScrollId());
        }
        etlProgressDO.setScrollTime(rangeScroll.getStartTime());
        etlProgressDO.setScrollEndTime(rangeScroll.getEndTime());
        etlProgressDO.setCreateTime(new DateTime());
        etlProgressDO.setUpdateTime(new DateTime());
        migrateScrollMapper.insertEtlProgress(etlProgressDO);

        return etlProgressDO;
    }

    /**
     * 更新数据迁移的进度明细信息
     *
     * @param rangeScroll 要检查的数据抽取模型
     * @param status      状态
     * @param errorMsg    异常信息
     */
    private EtlDirtyRecordDO updateEtlDirtyRecord(RangeScroll rangeScroll, Integer status, String errorMsg) {
        MigrateScrollMapper migrateScrollMapper = SpringContextUtil.getBean(MigrateScrollMapper.class);

        EtlDirtyRecordDO etlDirtyRecordDO = new EtlDirtyRecordDO();
        etlDirtyRecordDO.setLogicModel(rangeScroll.getTableName());
        etlDirtyRecordDO.setTicket(rangeScroll.getTicket());
        etlDirtyRecordDO.setCurTicketStage(rangeScroll.getCurTicketStage());
        etlDirtyRecordDO.setStatus(status);
        etlDirtyRecordDO.setErrorMsg(errorMsg);
        etlDirtyRecordDO.setUpdateTime(new DateTime());
        migrateScrollMapper.updateEtlDirtyRecord(etlDirtyRecordDO);
        return etlDirtyRecordDO;
    }

    /**
     * 更新 本次迁移数据的状态
     *
     * @param etlProgressDO 迁移记录
     * @param status        迁移状态
     * @param rangeScroll   要检查的数据抽取模型
     * @param errorMsg      异常信息
     */
    private void updateEtlDirtyRecord(EtlProgressDO etlProgressDO, Integer status, RangeScroll rangeScroll, String errorMsg) {
        //核对校验成功情况下，更新迁移记录
        if (errorMsg == null) {
            MigrateScrollMapper migrateScrollMapper = SpringContextUtil.getBean(MigrateScrollMapper.class);
            etlProgressDO.setUpdateTime(new DateTime());
            etlProgressDO.setCurTicketStage(rangeScroll.getCurTicketStage());
            etlProgressDO.setFinishRecord(etlProgressDO.getFinishRecord() + rangeScroll.getPageSize());
            etlProgressDO.setProgressType(1);
            // 更新本批次对应的迁移数据核对数据量
            migrateScrollMapper.updateEtlProgress(etlProgressDO);
        }
        // 更新迁移核对的明细
        updateEtlDirtyRecord(rangeScroll, status, errorMsg);
    }

    /**
     * 更新本次核对校验对应的迁移记录为核对校验完成
     *
     * @param queryEtlProgressMap 查询迁移记录的集合
     */
    public void updateEtlProgressCheckSuccess(Map<String, RangeScroll> queryEtlProgressMap) {
        MigrateScrollMapper migrateScrollMapper = SpringContextUtil.getBean(MigrateScrollMapper.class);

        Collection<RangeScroll> rangeScrolls = queryEtlProgressMap.values();
        for (RangeScroll rangeScroll : rangeScrolls) {
            EtlProgressDO etlProgressDO = migrateScrollMapper.queryEtlProgressCheck(rangeScroll);

            //判断是否所有批次都数据校验完成
            if (checkEtlProgressCheckSuccess(etlProgressDO, migrateScrollMapper)) {
                etlProgressDO.setUpdateTime(new DateTime());
                etlProgressDO.setStatus(EtlProgressStatus.CHECK_SUCCESS.getValue());
                migrateScrollMapper.updateEtlProgress(etlProgressDO);
            }
        }
    }

    /**
     * 判断迁移记录是否已经核对校验完成
     *
     * @param etlProgressDO       迁移记录
     * @param migrateScrollMapper 滚动迁移的Mapper组件
     * @return 是否核对校验完成
     */
    public boolean checkEtlProgressCheckSuccess(EtlProgressDO etlProgressDO, MigrateScrollMapper migrateScrollMapper) {
        if (etlProgressDO == null) {
            return false;
        }

        Integer startTime = Integer.valueOf(LocalDateTimeUtil.format(etlProgressDO.getScrollTime(), DatePattern.PURE_DATE_PATTERN));
        Integer endTime = Integer.valueOf(LocalDateTimeUtil.format(etlProgressDO.getScrollEndTime(), DatePattern.PURE_DATE_PATTERN));

        EtlStatisticalDO etlStatisticalDO = new EtlStatisticalDO();
        etlStatisticalDO.setDomain(etlProgressDO.getDomain());
        etlStatisticalDO.setLogicModel(etlProgressDO.getLogicModel());
        etlStatisticalDO.setStartTime(startTime);
        etlStatisticalDO.setEndTime(endTime);

        BigDecimal statisticalCount = migrateScrollMapper.getStatisticalCount(etlStatisticalDO);
        statisticalCount = statisticalCount == null ? BigDecimal.ZERO : statisticalCount;
        if (new BigDecimal(etlProgressDO.getFinishRecord().toString()).compareTo(statisticalCount) >= 0) {
            return true;
        }

        return false;
    }

}
